package cn.com.rocketmq.async;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;

/**
 * 异步发送
 * @author xuhaixiang
 * @date 2023-09-09
 */
public class Producer {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(cn.com.rocketmq.sync.Producer.class);


    public static void main(String[] args) {

        DefaultMQProducer producer = null;
        try {
            producer = new DefaultMQProducer("asyncProducer");
            producer.setNamesrvAddr("192.168.113.101:9876");
            producer.start();
            for (int i = 0; i < 5; i++) {
                String body = "Hello zhang " + i;
                //参数一：主题、参数二：过滤、参数三：消息内容
                Message message = new Message("rocketmq_asyncDemo","tag", body.getBytes("UTF-8"));
                //单向发送
                //producer.sendOneway(message);
                //异步发送
                producer.send(message, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        String msgId = sendResult.getMsgId();
                        SendStatus sendStatus = sendResult.getSendStatus();
                        logger.info("{}消息发送状态为{}", msgId, sendStatus);
                    }
                    @Override
                    public void onException(Throwable throwable) {
                        logger.error("消息发送失败", throwable);
                    }
                });
            }
        } catch (Exception e) {
            logger.error("生产者发送消息失败！" ,e);
        } finally {
            //异步发送不应该关闭
        }
    }
}
